package org.mq.mymq.broker.netty.broker;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.mq.mymq.broker.cache.CommonCache;
import org.mq.mymq.broker.event.model.*;
import org.mq.mymq.common.cache.BrokerServerSyncFutureManager;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.*;
import org.mq.mymq.common.enums.BrokerEventCode;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.EventBus;
import org.mq.mymq.common.event.model.Event;
import org.mq.mymq.common.remote.SyncFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;

/**
 * @ClassName BrokerServerHandler
 * @Description 处理Broker与生产者，消费者交互事件，Broker中master节点与Slave节点交互事件
 * 1. 处理生产者发送消息到broker信号，并发布生产消息事件
 * 2. 处理消费者拉取消息消费信号，发布消息消费事件，并监听消息消费成功ack信号
 * 3. 处理创建topic信号及处理创建成功信号，生成创建topic事件
 * 4. 接收Slave节点发送的开始同步消息信号并发布消息同步事件
 * 5. 接收slave节点同步消息信号成功信号，返回该成功信号给master节点
 * @Author jiarong_ye
 * @Date 2025/6/27 13:07
 * @Version 1.0
 */
public class BrokerServerHandler extends SimpleChannelInboundHandler<TcpMsg> {
    private static final Logger log = LoggerFactory.getLogger(BrokerServerHandler.class);
    private EventBus eventBus;

    public BrokerServerHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        this.eventBus.init();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        Event event = null;
        InetSocketAddress socketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
        if (BrokerEventCode.PUSH_MSG.getCode() == code) {
            MessageDTO messageDTO = JSON.parseObject(body, MessageDTO.class);
            PushMsgEvent pushMsgEvent = new PushMsgEvent();
            pushMsgEvent.setMessageDTO(messageDTO);
            event = pushMsgEvent;
            log.info("收到消息推送内容：{}，message is {}", new String(messageDTO.getBody()), JSON.toJSONString(messageDTO));
        } else if (BrokerEventCode.CONSUME_MSG.getCode() == code) {
            ConsumeMsgReqDTO consumeMsgReqDTO = JSON.parseObject(body, ConsumeMsgReqDTO.class);
            consumeMsgReqDTO.setIp(socketAddress.getAddress().getHostAddress());
            consumeMsgReqDTO.setPort(socketAddress.getPort());
            ConsumeMsgEvent consumeMsgEvent = new ConsumeMsgEvent();
            consumeMsgEvent.setConsumeMsgReqDTO(consumeMsgReqDTO);
            consumeMsgEvent.setMsgId(consumeMsgReqDTO.getMsgId());
            ctx.attr(AttributeKey.valueOf("consumer-reqId")).set(consumeMsgReqDTO.getIp() + ":" + consumeMsgReqDTO.getPort());
            event = consumeMsgEvent;
        } else if (BrokerEventCode.CONSUME_SUCCESS_MSG.getCode() == code) {
            ConsumeMsgAckReqDTO ackReqDTO = JSON.parseObject(body, ConsumeMsgAckReqDTO.class);
            ackReqDTO.setIp(socketAddress.getAddress().getHostAddress());
            ackReqDTO.setPort(socketAddress.getPort());
            ConsumeMsgAckEvent  consumeMsgAckEvent = new ConsumeMsgAckEvent();
            consumeMsgAckEvent.setConsumeMsgAckReqDTO(ackReqDTO);
            consumeMsgAckEvent.setMsgId(ackReqDTO.getMsgId());
            event = consumeMsgAckEvent;
        } else if (BrokerEventCode.CREATE_TOPIC.getCode() == code) {
            CreateTopicReqDTO createTopicReqDTO = JSON.parseObject(body, CreateTopicReqDTO.class);
            CreateTopicEvent createTopicEvent = new CreateTopicEvent();
            createTopicEvent.setCreateTopicReqDTO(createTopicReqDTO);
            createTopicEvent.setMsgId(createTopicReqDTO.getMsgId());
            event = createTopicEvent;
        } else if (BrokerEventCode.START_SYNC_MSG.getCode() == code) {
            StartSyncReqDTO startSyncReqDTO = JSON.parseObject(body, StartSyncReqDTO.class);
            StartSyncEvent startSyncEvent = new StartSyncEvent();
            startSyncEvent.setMsgId(startSyncReqDTO.getMsgId());
            event = startSyncEvent;
        } else if (BrokerResponseCode.CREATED_TOPIC_SUCCESS.getCode() == code) {
            // TODO 从节点返回创建TOPIC成功
            return;
        } else if (BrokerResponseCode.SLAVE_SYNC_RESP.getCode() == code) {
            SlaveSyncRespDTO slaveSyncRespDTO = JSON.parseObject(body, SlaveSyncRespDTO.class);
            SyncFuture slaveAckResp = BrokerServerSyncFutureManager.get(slaveSyncRespDTO.getMsgId());
            if (slaveAckResp != null) {
                slaveAckResp.setResponse(slaveSyncRespDTO);
            }
            return;
        }
        event.setChannelHandlerContext(ctx);
        eventBus.publish(event);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        log.error(cause.getMessage(), cause);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        super.channelActive(ctx);
        log.info("new connection is active");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        super.channelInactive(ctx);
        Object reqId = ctx.attr(AttributeKey.valueOf("consumer-reqId")).get();
        if (reqId == null) {
            return;
        }
        CommonCache.getConsumerInstancePool().removeInstancePool(String.valueOf(reqId));
    }
}
